// -----------------------------------------------------------------------
// <copyright file="GossipFoo.cs" company="Asynkron AB">
//      Copyright (C) 2015-2025 Asynkron AB All rights reserved
// </copyright>
// -----------------------------------------------------------------------


/*
   Member abc state:
       GossipState
           MemberState abc
               Key123 - sequence id 1
               Key456 - sequence id 2
   
           MemberState def
               Key123 - sequence id 1
               Key456 - sequence id 2
   
       committed offsets
           "abc.def" 1
   
   Member def state:
       GossipState
           MemberState abc
               Key123 - sequence id 1
               Key456 - sequence id 2
   
           MemberState def
               Key123 - sequence id 3
               Key456 - sequence id 2
   
       committed offsets
           "abc.def" 2
   
   
   gossip from def to abc
   scan all entries for all member states _except_ for abc (we shouldn´t send their state to them)
   find all entries that are higher than the committed offset for the member state
   send all entries to the target member
   
   committed offsets is local per member, meaning we will send excessive and needless data to other members
   maybe this could be improved by having each nodes committed offsets as part of the MemberState (?)   
 */
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Linq;
using System.Threading.Tasks;
using Google.Protobuf;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Extensions.Logging;
using Proto.Logging;

namespace Proto.Cluster.Gossip;

internal class Gossip
    : IGossip
{
    private static readonly ILogger Logger = Log.CreateLogger<Gossip>();
    private readonly ConsensusChecks _consensusChecks = new();
    private readonly Func<ImmutableHashSet<string>> _getMembers;
    private readonly bool _gossipDebugLogging;
    private readonly int _gossipFanout;
    private readonly int _gossipMaxSend;
    private readonly InstanceLogger? _logger;
    private readonly string _myId;
    private readonly IRandomProvider _random;
    private readonly MemberStateDeltaBuilder _memberStateDeltaBuilder;
    private ImmutableHashSet<string> _activeMemberIds = ImmutableHashSet<string>.Empty;
    private ImmutableDictionary<string, long> _committedOffsets = ImmutableDictionary<string, long>.Empty;
    private long _localSequenceNo;
    private Member[] _otherMembers = Array.Empty<Member>();
    private GossipState _state = new();

    public Gossip(string myId, int gossipFanout, int gossipMaxSend, InstanceLogger? logger,
        Func<ImmutableHashSet<string>> getMembers, bool gossipDebugLogging, IRandomProvider? randomProvider = null)
    {
        _myId = myId;
        _logger = logger;
        _getMembers = getMembers;
        _gossipDebugLogging = gossipDebugLogging;
        _gossipFanout = gossipFanout;
        _gossipMaxSend = gossipMaxSend;
        _random = randomProvider ?? new SystemRandomProvider();
        _memberStateDeltaBuilder = new MemberStateDeltaBuilder(myId, gossipMaxSend);
    }

    public Task UpdateClusterTopology(ClusterTopology clusterTopology)
    {
        var others = new List<Member>();
        var activeIds = ImmutableHashSet.CreateBuilder<string>();

        foreach (var m in clusterTopology.Members)
        {
            activeIds.Add(m.Id);

            if (m.Id != _myId)
            {
                others.Add(m);
            }
        }

        _otherMembers = others.ToArray();
        _activeMemberIds = activeIds.ToImmutable();

        Purge();
        SetState(GossipKeys.Topology, clusterTopology);

        return Task.CompletedTask;
    }

    public void AddConsensusCheck(string id, ConsensusCheck check)
    {
        _consensusChecks.Add(id, check);

        // Check when adding, if we are already consistent
        check.Check(_state, _activeMemberIds);
    }

    public void RemoveConsensusCheck(string id) => _consensusChecks.Remove(id);

    public GossipState GetStateSnapshot() => _state.Clone();

    public ImmutableDictionary<string, Any> GetState(string key)
    {
        var entries = ImmutableDictionary<string, Any>.Empty;

        foreach (var (memberId, memberState) in _state.Members)
        {
            if (memberState.Values.TryGetValue(key, out var value))
            {
                entries = entries.SetItem(memberId, value.Value);
            }
        }

        return entries;
    }

    public ImmutableList<GossipUpdate> ReceiveState(GossipState remoteState)
    {
        var (newState, updates, updatedKeys) = GossipStateManagement.MergeStates(_state, remoteState);

        if (updates.Count == 0)
        {
            return ImmutableList<GossipUpdate>.Empty;
        }

        if (_gossipDebugLogging)
        {
            Logger.LogInformation("ReceiveState: Gossip updates {Updates}", updates);
            Logger.LogInformation("ReceiveState: Old Gossip state {State}", _state);
            Logger.LogInformation("ReceiveState: New Gossip state {State}", newState);
            Logger.LogInformation("ReceiveState: Updated Keys {Keys}", updatedKeys);
        }

        _state = newState;

        CheckConsensus(updatedKeys);

        return updates.ToImmutableList();
    }

    public void SetState(string key, IMessage message)
    {
        var logger = _logger?.BeginMethodScope();
        _localSequenceNo = GossipStateManagement.SetKey(_state, key, message, _myId, _localSequenceNo);
        logger?.LogDebug("Setting state key {Key} - {Value} - {State}", key, message, _state);
        Logger.LogDebug("Setting state key {Key} - {Value} - {State}", key, message, _state);
        
        if (_gossipDebugLogging)
        {
            Logger.LogInformation("SetState: Gossip key {Key} - {Value} - {State}", key, message, _state);
        }

        if (!_state.Members.ContainsKey(_myId))
        {
            Logger?.LogCritical("State corrupt");
            logger?.LogCritical("State corrupt");
        }

        CheckConsensus(key);
    }

    public IEnumerable<(Member member, MemberStateDelta memberState)> SendState()
    {
        var sends = new List<(Member member, MemberStateDelta memberState)>();

        try
        {
            _logger?.BeginMethodScope();

            foreach (var member in _otherMembers)
            {
                GossipStateManagement.EnsureMemberStateExists(_state, member.Id);
            }

            var randomMembers = _otherMembers.OrderByRandom(_random).ToArray();

            var fanoutCount = 0;

            if (_gossipDebugLogging)
            {
                var ids = randomMembers.Select(m => m.Id).ToArray();
                Logger.LogInformation("SendState: Gossip to members {Members}", ids);
            }

            foreach (var member in randomMembers)
            {
                var memberState = GetMemberStateDelta(member.Id);

                if (!memberState.HasState)
                {
                    continue;
                }

                sends.Add((member, memberState));

                fanoutCount++;

                if (fanoutCount == _gossipFanout)
                {
                    break;
                }
            }
        }
        catch (Exception x)
        {
            Logger.LogError(x, "SendState failed");
        }

        return sends;
    }

    public MemberStateDelta GetMemberStateDelta(string targetMemberId)
    {
        var (state, pendingOffsets, hasState) =
            _memberStateDeltaBuilder.Build(_state, targetMemberId, _committedOffsets, _random);

        return new MemberStateDelta(targetMemberId, hasState, state, () => CommitPendingOffsets(pendingOffsets));
    }

    public ImmutableDictionary<string, GossipKeyValue> GetStateEntry(string key)
    {
        var entries = ImmutableDictionary<string, GossipKeyValue>.Empty;

        foreach (var (memberId, memberState) in _state.Members)
        {
            if (memberState.Values.TryGetValue(key, out var value))
            {
                entries = entries.SetItem(memberId, value);
            }
        }

        return entries;
    }

    private void CheckConsensus(string updatedKey)
    {
        foreach (var consensusCheck in _consensusChecks.GetByUpdatedKey(updatedKey))
        {
            consensusCheck.Check(_state, _activeMemberIds);
        }
    }

    private void CheckConsensus(IEnumerable<string> updatedKeys)
    {
        foreach (var consensusCheck in _consensusChecks.GetByUpdatedKeys(updatedKeys))
        {
            consensusCheck.Check(_state, _activeMemberIds);
        }
    }

    private void Purge()
    {
        //find all members that have sent topology
        var members = _getMembers();
        
        //purge member states
        foreach (var memberId in _state.Members.Keys.ToArray())
        {
            if (!members.Contains(memberId))
            {
                _state.Members.Remove(memberId);
            }
        }

        //purge committed offsets
        foreach (var x in _committedOffsets.Keys.ToArray())
        {
            var parts = x.Split(".");
            var from = parts[0];
            var to = parts[1];
            if (!members.Contains(from) || !members.Contains(to))
            {
                _committedOffsets = _committedOffsets.Remove(x);
            }
        }
    }

    private void CommitPendingOffsets(ImmutableDictionary<string, long> pendingOffsets)
    {
        foreach (var (key, sequenceNumber) in pendingOffsets)
        {
            if (!_committedOffsets.TryGetValue(key, out var current) || current < sequenceNumber)
            {
                _committedOffsets = _committedOffsets.SetItem(key, sequenceNumber);
            }
        }
    }
}